package com.saga.energy.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.saga.energy.bean.EnergyHistory;
import com.saga.energy.sink.Energy_15_fjd_history_sink;
import com.saga.energy.sink.Energy_15_his_list_sink;
import com.saga.energy.sink.Energy_15_history_sink;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Properties;

public class KafkaToMysqlBatch {

    public static void main(String[] args) throws Exception {



        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop01:9092,hadoop02:9092,hadoop03:9092");
        properties.setProperty("group.id", "energy-03");
        properties.setProperty("auto.offset.reset", "earliest");

        DataStreamSource<String> energy_15_min = env.addSource(new FlinkKafkaConsumer<>("ENERGY_15_MIN", new SimpleStringSchema(), properties));
//        DataStreamSource<String> energy_15_min_fjd = env.addSource(new FlinkKafkaConsumer<>("ENERGY_15_MIN_FJD", new SimpleStringSchema(), properties));

//        JDBCOutputFormat

        SingleOutputStreamOperator<EnergyHistory> energy_15_minStream = energy_15_min.flatMap(new FlatMapFunction<String, EnergyHistory>() {
            @Override
            public void flatMap(String s, Collector<EnergyHistory> collector) {

                JSONObject jsonObject = JSON.parseObject(s);
                jsonObject.put("dt", jsonObject.getString("data_time").substring(0, 8));
//                System.out.println(jsonObject.toJSONString());
                EnergyHistory energyHistory = JSON.parseObject(jsonObject.toJSONString(), EnergyHistory.class);

                collector.collect(energyHistory);
            }
        });

        KeyedStream<EnergyHistory, String> energyHistoryStringKeyedStream = energy_15_minStream.keyBy(new KeySelector<EnergyHistory, String>() {
            @Override
            public String getKey(EnergyHistory energyHistory) throws Exception {
                String o = energyHistory.getBuilding() + energyHistory.getFuncid() + energyHistory.getMeter();
                return o;
            }
        });

        WindowedStream<EnergyHistory, String, TimeWindow> energy_15_min_window =
                energyHistoryStringKeyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(2)));

        SingleOutputStreamOperator<List<EnergyHistory>> energy_15_min_process = energy_15_min_window.process(new ProcessWindowFunction<EnergyHistory, List<EnergyHistory>, String, TimeWindow>() {
            @Override
            public void process(String s, ProcessWindowFunction<EnergyHistory, List<EnergyHistory>, String, TimeWindow>.Context context,
                                Iterable<EnergyHistory> iterable,
                                Collector<List<EnergyHistory>> collector) throws Exception {
                Iterator<EnergyHistory> iterator = iterable.iterator();
                List<EnergyHistory> energyHistories = new ArrayList<>();
                while (iterator.hasNext()){
                    EnergyHistory next = iterator.next();
                    energyHistories.add(next);
                }
                collector.collect(energyHistories);
            }
        });


        energy_15_min_process.addSink(new Energy_15_his_list_sink());
//        energy_15_minStream.print();

        env.execute("energy_15_min_his : ");

    }
}
